package com.lry.basic.mq;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author:刘仁有
 * @desc:
 * @email:953506233@qq.com
 * @data:2019/6/19
 */
public class MessageQueue<T> {
    Message<T> items[];
    int putIndex;
    int takeIndex;
    int count;
    static final int SIZE = 10;
    Lock lock;
    Condition notFull;//生产者
    Condition notEmpty;//消费者
    public MessageQueue(){
        init();
    }

    private void init() {
        items = new Message[SIZE];
        lock = new ReentrantLock();
        notFull = lock.newCondition();
        notEmpty = lock.newCondition();
    }

    public void enqueue(Message<T> msg){
        try {
            lock.lock();
            if(count==SIZE){
                notFull.await();
            }
            items[putIndex] = msg;
            putIndex = (++putIndex==SIZE)?0:putIndex;
            count++;
            //生产完通知消费者消费
            notEmpty.signalAll();
        }catch (InterruptedException e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    public Message deque(){
        Message msg = null;
        try {
            lock.lock();
            if(count==0){
                notEmpty.await();
            }
            msg = items[takeIndex];
            items[takeIndex] = null;
            takeIndex = (++takeIndex==SIZE)?0:takeIndex;
            count--;
            //消费完通知生产者生产
            notFull.signalAll();
        }catch (InterruptedException e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        return msg;
    }

}
